package io.mqttpush.mqttserver.config;

import io.mqttpush.common.events.CommonTopics;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author tianzhenjiu
 * @date 2020/4/27 16:47
 */
public class TopicManager {

    Logger logger= LoggerFactory.getLogger(getClass());

    /**
     * 单个主题->channels
     */
   final   ConcurrentHashMap<String, Set<Channel>> subscriptions = new ConcurrentHashMap<> ();


    /**
     *
     * 多个主题->channels
     */
    final   ConcurrentHashMap<String, Set<Channel>> mutilSubscriptions = new ConcurrentHashMap<> ();


   final  Map<String, TopicInfo> topicInfos = new ConcurrentHashMap<> ();



   final String  endStart="*";


   private static TopicManager topicManager=new TopicManager();



   public  TopicManager(){

       addNewTopic(CommonTopics.eventTopic,getClass().getName());

       addNewTopic(CommonTopics.comTopic,getClass().getName());
   }





    /**
     *单例模式获取实例
     *
     * @return
     */
    public  static  TopicManager getInstance(){

        return topicManager;
    }


    public  void  subscribe(String topic, Channel channel, MqttQoS qos){


        logger.info("channel subscribe topic->{},{}",channel,topic);

        DeviceManager deviceManager=DeviceManager.getDeviceManager();
        Map<String,MqttQoS> mqttQoSMap=channel.attr(deviceManager.channelTopics).get();
        if(mqttQoSMap!=null){
            mqttQoSMap.put(topic,qos);
        }
        /**
         *
         * 包含星号的认为是通配主题
         */
        if(topic.endsWith(endStart)){

            if(topic.length()<=endStart.length()){
                logger.info("不合法的topic{}",topic);
                return;
            }
            topic=topic.substring(0,topic.length()-endStart.length());
            subscribe(topic,channel,mutilSubscriptions);
        }
        else{
            subscribe(topic,channel,subscriptions);
        }

    }


    public  void  subscribe(String topic, Channel channel, ConcurrentHashMap<String, Set<Channel>>subscriptions){



        Set<Channel> channels=null;


        if(!subscriptions.containsKey(topic)){

            if(!topicInfos.containsKey(topic)){

                addNewTopic(topic,channel.id().asShortText());
            }

            subscriptions.put(topic,  channels=new HashSet<>(16));
        }else{
            channels=subscriptions.get(topic);
        }

        channels.add(channel);


    }

    public  void  unSubscribe(String topic, Channel channel){



        logger.info("channel unsubscribe topic->{},{}",channel,topic);

        DeviceManager deviceManager=DeviceManager.getDeviceManager();
        Map<String,MqttQoS> mqttQoSMap=channel.attr(deviceManager.channelTopics).get();
        if(mqttQoSMap!=null){
            mqttQoSMap.remove(topic);
        }
        /**
         *
         * 包含星号的认为是通配主题
         */

        if(topic.endsWith(endStart)){

            if(topic.length()<=endStart.length()){
                logger.info("不合法的topic{}",topic);
                return;
            }

            topic=topic.substring(0,topic.length()-endStart.length());
            unSubscribe(topic,channel,mutilSubscriptions);
        }
        else{
            unSubscribe(topic,channel,subscriptions);
        }
    }

    public  void  unSubscribe(String topic, Channel channel,ConcurrentHashMap<String, Set<Channel>> subscriptions){



        Set<Channel> channels=null;


        if(!subscriptions.containsKey(topic)){

            logger.info("the topic you  will unsubscribe {} is not exists",topic);
            return;

        }

        channels=subscriptions.get(topic);


        /**
         *
         * 如果订阅topc的channels空了，并且topicInfos不包含这个topic了，就要清楚topic的订阅
         */
        if(channels==null||channels.isEmpty()){
            if(!topicInfos.containsKey(topic)){
                subscriptions.remove(topic);
            }
           return;
        }

        channels.remove(channel);


        logger.info("the channel {} unsubscribe the topic {}",channel,topic);

        if(channels==null||channels.isEmpty()){
            if(!topicInfos.containsKey(topic)){
                subscriptions.remove(topic);
            }
            return;
        }


    }


    /**
     *
     * 查询出所有匹配指定topic的channel
     * @param topic
     * @return
     */
    public Set<Channel> topicChannels(String topic){



        Set<Channel> routedChannels=new HashSet<>();


        Set<Channel> channels=new HashSet<>(4);


        if(subscriptions.containsKey(topic)){
            channels.addAll(subscriptions.get(topic));
        }


        mutilSubscriptions.forEach((k,v)->{
            if(topic.startsWith(k)){
                channels.addAll(v);
            }
        });

        channels.forEach(channel -> {

            if(channel.isActive()){
                routedChannels.add(channel);
            }


        });


        return routedChannels;

    }


    public  void addNewTopic(String topic,String fromId){

        TopicInfo topicInfo=new TopicInfo();
        topicInfo.createrId=fromId;
        topicInfo.createTime=new Date();
        topicInfo.topicName=topic;

        topicInfos.put(topic,topicInfo);

    }



    public  void deleteExistsTopic(String topic,String fromId){


        if(!topicInfos.containsKey(topic)){
            logger.info("the will  delete  topic {} is not exists",topic);
            return;
        }

        topicInfos.remove(topic);
        if(subscriptions.containsKey(topic)){
            Set<Channel>  channels=subscriptions.get(topic);
            if(channels.isEmpty()){
                subscriptions.remove(topic);
            }
        }


    }


}
